package test.org.offsync.async

import java.net.InetSocketAddress
import java.net.Socket
import java.net.SocketAddress
import java.util.concurrent.CyclicBarrier
import java.util.concurrent.atomic.AtomicInteger
import org.junit.runner.RunWith
import org.offsync.old.async.AsyncResult
import org.offsync.old.async.AsyncReturn
import org.offsync.old.async.AsyncService
import org.offsync.old.async.AsyncServerSocket
import org.offsync.old.async.AsyncSocket
import org.offsync.util.Disposer
import org.offsync.util.Tasks
import org.offsync.util.Time
import org.specs2._
import org.offsync.util.DisposeState
import scala.collection.{mutable => mutable}

@RunWith(classOf[org.specs2.runner.JUnitRunner])
class SpecAsyncService extends org.specs2.mutable.SpecificationWithJUnit {
  "New AsyncService object can dispose" should {
    val asyncService = new AsyncService()
    asyncService.waitDisposed(Time.now)
    asyncService.disposeState must_== DisposeState.Live
    asyncService.dispose()
    1 must_== 1
  }

  "AsyncService object can waitDisposed" in {
    val asyncService = new AsyncService()
    var counter = 0
    val thread = new Thread() {
      override def run(): Unit = {
        Thread.sleep(10)
        counter *= 2
        asyncService.dispose()
      }
    }
    thread.start()
    asyncService.waitDisposed()
    counter += 1
    thread.join()
    asyncService.disposeState must_== DisposeState.Disposed
  }
  
  "AsyncService object can dispose when accepting in AsyncServerSocket object" in {
    val barrier = new CyclicBarrier(2)
    val disposer = new Disposer()
    try {
      val asyncService = disposer.disposes(new AsyncService())
      val asyncServerSocket = disposer.disposes(AsyncServerSocket.open())
      
      asyncServerSocket.accept(asyncService, null, 9876) { result =>
        val socketChannel = result.value
        disposer.disposes(socketChannel)
        barrier.await()
        AsyncReturn
      }
      val socket = new Socket()
      
      try {
        socket.connect(new InetSocketAddress("localhost", 9876))
        Thread.sleep(100)
      } finally {
        socket.close()
      }
      barrier.await()
    } finally {
      disposer.dispose()
    }
    
    disposer.disposeState must_== DisposeState.Disposed
  }
  
  "AsyncService object can dispose when reading in AsyncServerSocket object" in  {
    val barrier = new CyclicBarrier(2)
    val disposer = new Disposer()
    try {
      val asyncService = disposer.disposes(new AsyncService())
      val asyncServerSocket = disposer.disposes(AsyncServerSocket.open())
      
      asyncServerSocket.accept(asyncService, null, 9876) { result =>
        val socketChannel = result.value
        socketChannel.read(asyncService, 1) { result =>
          val maybeArray = result.value
          maybeArray must_== None
          barrier.await()
          AsyncReturn
        }
      }
      val socket = new Socket()
      
      try {
        socket.connect(new InetSocketAddress("localhost", 9876))
        
        Thread.sleep(100)
      } finally {
        socket.close()
      }
      barrier.await()
    } finally {
      disposer.dispose()
    }
    
    disposer.disposeState must_== DisposeState.Disposed
  }

  "AsyncService can read" in  {
    var assertions = new Tasks
    val barrier = new CyclicBarrier(2)
    val disposer = new Disposer()
    try {
      val asyncService = disposer.disposes(new AsyncService())
      val asyncServerSocket = disposer.disposes(AsyncServerSocket.open())
      
      asyncServerSocket.accept(asyncService, null, 9876) { result =>
        val socketChannel = result.value
        socketChannel.read(asyncService, 1) { result =>
          assertions.add { result.value.toList(0).toList must_== List[Byte](42) }
          socketChannel.read(asyncService, 1) { result =>
            assertions.add { result.value.toList(0).toList must_== List[Byte](43) }
            barrier.await()
            AsyncReturn
          }
        }
      }
      val socket = new Socket()
      
      try {
        socket.connect(new InetSocketAddress("localhost", 9876))
        Thread.sleep(100)
        socket.getOutputStream().write(Array[Byte](42))
        Thread.sleep(100)
        socket.getOutputStream().write(Array[Byte](43))
        
        Thread.sleep(100)
      } finally {
        socket.close()
      }
      barrier.await()
    } finally {
      disposer.dispose()
    }
    
    disposer.disposeState must_== DisposeState.Disposed
    assertions.run()
    1 must_== 1
  }

  "AsyncService can read and write" in  {
    var assertions = new Tasks
    val barrier = new CyclicBarrier(2)
    val disposer = new Disposer()
    try {
      val asyncService = disposer.disposes(new AsyncService())
      val asyncServerSocket = disposer.disposes(AsyncServerSocket.open())
      
      asyncServerSocket.accept(asyncService, null, 9876) { result =>
        val socketChannel = result.value
        socketChannel.read(asyncService, 1) { result =>
          assertions.add(result.value.toList(0).toList must_== List[Byte](42))
          socketChannel.read(asyncService, 1) { result =>
            assertions.add(result.value.toList(0).toList must_== List[Byte](44))
            barrier.await()
            AsyncReturn
          }
        }
      }
      
      val writeSocketChannel = AsyncSocket.open(new InetSocketAddress("localhost", 9876))
      
      try {
        writeSocketChannel.write(asyncService, Array[Byte](42)) { result =>
          writeSocketChannel.write(asyncService, Array[Byte](44)) { result =>
            AsyncReturn
          }
        }
        
        Thread.sleep(100)
      } finally {
        writeSocketChannel.dispose()
      }
      barrier.await()
    } finally {
      disposer.dispose()
    }
    
    disposer.disposeState must_== DisposeState.Disposed
    assertions.run()
    1 must_== 1
  }
}
